use std::any::Any;
use std::sync::Arc;

use hashbrown::HashMap;
use parking_lot::Mutex;

// 单飞模式, 模拟 go 的 singleflight 写的
// 用于避免重复请求, 当有相同的请求时, 直接返回之前的结果
// 适用于耗时长的请求, 如网络请求, 数据库查询等
// 需要拥有多线程的内部可变性, 所以需要 Arc + Mutex
// 需要支持不同的类型分发到不同的线程, 所以要用 Arc<dyn Any + Send + Sync>
// type Call<T: Any + Send + Sync> = Arc<Mutex<Arc<T>>>;
type OptionResultAny = Option<Result<Option<Box<dyn Any + Send + Sync>>, Box<dyn Any + Send + Sync>>>;

#[derive(Debug, Default, Clone)] // Group 是需要给到不同的线程的, 所以需要 Clone
pub struct Group {
    // 这里加个 Arc + Mutex 包装一下, 使得内部可变性可以被多个线程访问
    m: Arc<Mutex<HashMap<String, Arc<Mutex<OptionResultAny>>>>>,
}

// impl Default for Group {
//     fn default() -> Self {
//         Self {
//             m: Arc::new(Mutex::new(HashMap::<
//                 String,
//                 Arc<Mutex<OptionArcAny>>,
//             >::new())),
//         }
//     }
// }

impl Group {
    pub fn work<T, F, E>(&self, key: &str, func: F) -> Result<Option<T>, E>
        where
            T: Clone + Any + Send + Sync,
            E: Clone + Any + Send + Sync,
            F: FnOnce() -> Result<Option<T>, E>,
    {
        // 加大锁, 避免并发访问
        let mut m = self.m.lock();
        // 如果任务 key 已存在, 获取包裹结果的锁
        if let Some(c) = m.get(key) {
            // 拷贝一下, 因为要释放大锁
            let c = c.clone();
            // 释放大锁
            drop(m);
            // 加锁等待, 返回之前的结果并释放锁
            let mut guard = c.lock();
            let res = guard.take().unwrap();

            let ret = res
                .map(|x| x.map(|v| v.downcast::<T>().unwrap()))
                .map_err(|e| e.downcast::<E>().unwrap());
            let res = ret.clone();
            guard.replace(res.map(|x| x.map(|v| v as _)).map_err(|x| x as _));
            drop(guard);
            return ret.map(|x| x.map(|v| *v)).map_err(|e| *e);
        }

        // 创建一个 call, 用来存储结果
        let call = Arc::new(Mutex::new(None));
        // 复制一下, 因为要插入到 map 中释放大锁
        let call_clone = call.clone();
        // 写入 map, 其他相同任务看到这个 key 后, 不会再执行 func 函数
        m.insert(key.to_owned(), call_clone);
        // 加锁, 阻止其他相同任务在结果写入前访问结果
        let mut call_guard = call.lock();
        // 释放大锁. 这把锁不能长时间拥有, 否则会影响其他任务执行
        drop(m);

        // 执行任务, 获取任务返回结果
        let res = match func() {
            Ok(v) => match v {
                Some(x) => Ok(Some(Box::new(x))),
                None => Ok(None),
            },
            Err(e) => { Err(Box::new(e)) }
        };
        // 将结果写入内部可变性的 Mutex 中
        call_guard.replace(res.clone().map(|x| x.map(|v| v as _)).map_err(|x| x as _));
        // 释放锁, 让其他相同的任务可以访问结果
        drop(call_guard);

        // 加大锁删除 map 中的 key, 避免内存泄漏, 并释放锁.
        // 在这之后拿到大锁的任务, 需要重新执行 func 函数
        self.m.lock().remove(key).unwrap();
        match res {
            Ok(v) => match v {
                Some(x) => Ok(Some(*x)),
                None => Ok(None),
            },
            Err(e) => Err(*e),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::thread;
    use std::thread::sleep;
    use std::time::Duration;

    use crate::single_flight_box::Group;

    #[derive(Clone)]
    struct A {
        value: i32,
    }

    #[test]
    fn test_single_flight() {
        use std::sync::mpsc;
        let count = 200;
        let (tx, rx) = mpsc::channel::<()>();
        let g: Group = Group::default();
        for i in 0..count {
            let g = g.clone();
            let tx = tx.clone();
            thread::spawn(move || {
                let (key, value) = if i % 2 == 0 {
                    ("aaa", 121)
                } else {
                    ("bbb", 111)
                };
                let res: Result<Option<i32>, String> = g.work(key, || {
                    sleep(Duration::from_millis(300));
                    Ok(Some(value))
                });
                assert_eq!(res.unwrap().unwrap(), value);
                drop(tx);
            });
        }

        for i in 0..count {
            let g = g.clone();
            let tx = tx.clone();
            thread::spawn(move || {
                let (key, value) = if i % 2 == 0 {
                    ("ccc", "121".to_owned())
                } else {
                    ("ddd", "111".to_owned())
                };
                let value2 = value.clone();
                let res: Result<Option<String>, String> = g.work(key, || {
                    sleep(Duration::from_millis(300));
                    Ok(Some(value2))
                });
                assert_eq!(res.unwrap().unwrap(), value);
                drop(tx);
            });
        }

        for i in 0..count {
            let g = g.clone();
            let tx = tx.clone();
            thread::spawn(move || {
                let (key, value) = if i % 2 == 0 {
                    ("eee", A { value: 121 })
                } else {
                    ("fff", A { value: 111 })
                };
                let value2 = value.clone();
                let res: Result<Option<A>, String> = g.work(key, || {
                    sleep(Duration::from_millis(300));
                    Ok(Some(value2))
                });
                assert_eq!(res.unwrap().unwrap().value, value.value);
                drop(tx);
            });
        }

        drop(tx);
        while rx.recv().is_ok() {}
    }
}
